package com.bw.app.function;

import com.alibaba.fastjson.JSONObject;
import com.bw.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

/**
 * @author whd
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        //进PhoenixConnection源码看 autoCommit
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    //value:{"sinkTable":"dim_base_trademark","database":"gmall-2021-flink","before":{},"after":{"id":13,"tm_name":"whd"},"type":"insert","tableName":"base_trademark"}
    //phoenix写的SQL语句:upsert into db.tn(id,tm_name) values (...,...)
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        PreparedStatement preparedStatement = null;
        try {
            //获取SQL语句
            String upsertSql = getUpsertSql(value.getString("sinkTable"),value.getJSONObject("after"));
            System.out.println(upsertSql);
            //预编译SQL
            preparedStatement = connection.prepareStatement(upsertSql);
            //执行插入操作
            preparedStatement.execute();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (preparedStatement != null){
                preparedStatement.close();
            }
        }

    }

    //SQL语句:upsert into db.tn(id,tm_name) values ('...','...','...','...')
    public static String getUpsertSql(String sinkTable, JSONObject after){
        Set<String> keySet = after.keySet();
        Collection<Object> values = after.values();
        return "upsert into "+GmallConfig.HBASE_SCHEMA+"."+sinkTable+"("+ StringUtils.join(keySet,",") + ")" + "values ('"+StringUtils.join(values,"','") + "')";
    }
}
